package com.witsight.cassandra.dao.impl;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cassandra.core.ResultSetExtractor;
import org.springframework.cassandra.core.RowCallback;
import org.springframework.cassandra.core.RowCallbackHandler;
import org.springframework.dao.DataAccessException;
import org.springframework.data.cassandra.core.CassandraConverterRowCallback;
import org.springframework.data.cassandra.core.CassandraOperations;
import org.springframework.data.cassandra.repository.query.CassandraEntityInformation;
import org.springframework.data.cassandra.repository.support.SimpleCassandraRepository;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.Pageable;
import org.springframework.util.Assert;

import com.datastax.driver.core.ResultSet;
import com.datastax.driver.core.Row;
import com.datastax.driver.core.exceptions.DriverException;
import com.witsight.cassandra.dao.BaseCassandraRepository;
import com.witsight.cassandra.dao.CassandraCommonRepository;
import com.witsight.cassandra.dao.NameQueriesRepository;
import com.witsight.cassandra.dao.PageCassandraOperations;
import com.witsight.cassandra.util.CqlBuilder;
import com.witsight.platform.dao.CommonRepository;
import com.witsight.platform.model.BoEntity;
import com.witsight.platform.model.IBaseObject;
import com.witsight.platform.util.lang.BeanUtil;

import joinery.DataFrame;

/**   
 * 说明：动态Cassandra仓库
 * @Title: DynamicCassandraRepository.java 
 * @Package com.witsight.cassandra.dao 
 * @See: {@link SimpleCassandraRepository}
 * 			{@link BaseCassandraRepository}
 * Copyright: Copyright (c) 2017
 * Company:sany huax witsight team by product
 * @author: penght  
 * @date: 2017年9月27日 上午9:32:35 
 * @version: V1.0
 *
 */
public class DynamicCassandraRepository<T, ID extends Serializable> extends SimpleCassandraRepository<T, ID> implements BaseCassandraRepository<T, ID> {
	/**
     * Logger
     */
    private static final Logger log = LoggerFactory.getLogger(DynamicCassandraRepository.class);
	private final NameQueriesRepository nameQueries;
	private final PageCassandraOperations operations;
	private final CassandraCommonRepository commonRepository;
	private final CassandraEntityInformation<T, ID> entityInformation;
	/**
	 * 
	 * 说明：构造函数 
	 * @Title:DynamicCassandraRepository.java
	 * @See: {@link DynamicCassandraRepository}
	 * @param metadata CassandraEntityInformation
	 * @param operations CassandraOperations
	 * @date 2017年9月27日 下午2:15:20
	 */
	public DynamicCassandraRepository(CassandraEntityInformation<T, ID> metadata, CassandraOperations operations, 
			NameQueriesRepository nameQueries,CommonRepository commonRepository) {
		super(metadata, operations);
		this.entityInformation = metadata;
		Assert.notNull(nameQueries, "NameQueriesRepository must not be null");
		this.nameQueries = nameQueries;
		Assert.isInstanceOf(PageCassandraOperations.class, operations, "operations must be PageCassandraOperations type.");
		this.operations = (PageCassandraOperations) operations;
		Assert.isInstanceOf(CassandraCommonRepository.class, commonRepository, "CommonRepository must be CassandraCommonRepository type.");
		this.commonRepository = (CassandraCommonRepository) commonRepository;
	}
	
	@Override
	public List<T> findAllByParams(String queryName, IBaseObject params) {
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName,bparam);
		Assert.hasText(query, "query must not be null");
		final List<T> resluts = new ArrayList<T>();
		this.operations.query(query, bparam, new RowCallbackHandler() {

			@Override
			public void processRow(Row row) throws DriverException {
				RowCallback<T> converter = new CassandraConverterRowCallback<T>(operations.getConverter(), entityInformation.getJavaType());
				resluts.add(converter.doWith(row));
			}
			
		});
		return resluts;
	}
	
	@Override
	public List<String> findBoEntitysByParams(String queryName, IBaseObject params) {
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName,bparam);
		Assert.hasText(query, "query must not be null");
		final List<String> resluts = new ArrayList<String>();
		this.operations.query(query, bparam, new RowCallbackHandler() {

			@Override
			public void processRow(Row row) throws DriverException {
				RowCallback<String> converter = new CassandraConverterRowCallback<String>(operations.getConverter(), String.class);
				resluts.add(converter.doWith(row));
			}
			
		});
		return resluts;
	}

	@Override
	public T findOneByParams(String queryName, IBaseObject params) {
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName,bparam);
		Assert.hasText(query, "query must not be null");
		return this.operations.query(query, bparam, new ResultSetExtractor<T>() {

			@Override
			public T extractData(ResultSet rs) throws DriverException, DataAccessException {
				Row row = rs.one();
				if (row == null) {
					return null;
				}
				RowCallback<T> converter = new CassandraConverterRowCallback<T>(operations.getConverter(), entityInformation.getJavaType());
				return converter.doWith(row);
			}

		});
	}

	@Override
	public String findBoEntityByParams(String queryName, IBaseObject params) {
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName,bparam);
		Assert.hasText(query, "query must not be null");
		return this.operations.query(query, bparam, new ResultSetExtractor<String>() {

			@Override
			public String extractData(ResultSet rs) throws DriverException, DataAccessException {
				Row row = rs.one();
				if (row == null) {
					return null;
				}
				RowCallback<String> converter = new CassandraConverterRowCallback<String>(operations.getConverter(), String.class);
				return converter.doWith(row);
			}

		});
	}
	
	@Override
	public Page<T> findPageByParams(String queryName, IBaseObject params, Pageable pageable) {
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName,bparam);
		Assert.hasText(query, "query must not be null");
		// 转化为统计cql
		String countCql = CqlBuilder.createCountQueryFor(query);
		long count = this.operations.selectCount(countCql);
		if (count <= 0) {
			return new PageImpl<T>(Collections.<T> emptyList());
		}
		List<T> list = this.operations.queryPage(query, bparam, pageable, new RowCallback<T>() {
			@Override
			public T doWith(Row row) {
				RowCallback<T> converter = new CassandraConverterRowCallback<T>(operations.getConverter(), entityInformation.getJavaType());
				return converter.doWith(row);
			}
		});
		return new PageImpl<T>(list, pageable, count);
	}

	@Override
	public <O> List<O> findSqlAllByParams(String queryName, IBaseObject params) {
		return commonRepository.findSqlAllByParams(queryName, params);
	}

	@Override
	public <O> O findSqlOneByParams(String queryName, IBaseObject params) {
		return commonRepository.findSqlOneByParams(queryName, params);
	}

	@Override
	public <O> Page<O> findSqlPageByParams(String queryName, IBaseObject params, Pageable pageable) {
		return commonRepository.findSqlPageByParams(queryName, params, pageable);
	}

	@Override
	public Page<String> findPageBoEntityByParams(String queryName, IBaseObject params, Pageable pageable) {
		BoEntity bparam= BeanUtil.object2Maps(params);
		String query = nameQueries.getQuery(queryName,bparam);
		Assert.hasText(query, "query must not be null");
		// 转化为统计cql
		String countCql = CqlBuilder.createCountQueryFor(query);
		long count = this.operations.selectCount(countCql);
		if (count <= 0) {
			return new PageImpl<String>(Collections.<String> emptyList());
		}
		List<String> list = this.operations.queryPage(query, bparam, pageable, new RowCallback<String>() {
			@Override
			public String doWith(Row row) {
				RowCallback<String> converter = new CassandraConverterRowCallback<String>(operations.getConverter(), String.class);
				return converter.doWith(row);
			}
		});
		return new PageImpl<String>(list, pageable, count);
	}

	@Override
	public String findPageWithJson(String queryName, IBaseObject params, Pageable pageable) {
		return commonRepository.findPageWithJson(queryName, params, pageable);
	}

	@Override
	public String findAllWithJson(String queryName, IBaseObject params) {
		return commonRepository.findAllWithJson(queryName, params);
	}

	@Override
	public DataFrame<Object> findDataFramesByParams(String queryName, IBaseObject params) {
		// TODO Auto-generated method stub
		return commonRepository.findDataFramesByParams(queryName, params);
	}
}
